package cn.sciento.transfer.infra.rocketmq.config;

import java.util.List;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.tools.command.SubCommandException;
import cn.sciento.transfer.infra.rocketmq.utils.RocketmqClientUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OffsetInitCapableRemoteStore {
  private static final Logger LOGGER = LoggerFactory.getLogger(OffsetInitCapableRemoteStore.class);
  
  public static void notifyRemoteServerUpdateOffset(String nameSrv, String consumerGroup, String topic, Long offset) {
    try {
      RocketmqClientUtils.createTopic(nameSrv, topic);
      List<MessageQueue> messageQueues = RocketmqClientUtils.queryQueueListByTopic(nameSrv, topic);
      for (MessageQueue messageQueue : messageQueues)
        RocketmqClientUtils.resetConsumerOffset(nameSrv, consumerGroup, topic, messageQueue, offset); 
      if (LOGGER.isDebugEnabled())
        LOGGER.debug("update offset scs , offset = {}", offset); 
    } catch (SubCommandException e) {
      LOGGER.error(e.getMessage(), (Throwable)e);
    } 
  }
}
